package com.atguigu.day09;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Timestamp;
import java.util.List;
import java.util.Map;

//CEP迟到事件处理
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<OrderEvent> streamSource = env.addSource(new SourceFunction<OrderEvent>() {
            @Override
            public void run(SourceContext<OrderEvent> ctx) throws Exception {
                OrderEvent o1 = new OrderEvent("order-1", "create", 1000L);
                ctx.collectWithTimestamp(o1, o1.timestamp);
                Thread.sleep(1000L);
                OrderEvent o2 = new OrderEvent("order-2", "create", 2000L);
                ctx.collectWithTimestamp(o2, o2.timestamp);
                Thread.sleep(1000L);
                OrderEvent o3 = new OrderEvent("order-1", "pay", 3000L);
                ctx.collectWithTimestamp(o3, o3.timestamp);
                Thread.sleep(1000L);
                ctx.emitWatermark(new Watermark(7000L));
                Thread.sleep(1000L);
                OrderEvent o4 = new OrderEvent("order-2", "pay", 8000L);
                ctx.collectWithTimestamp(o4, o4.timestamp);
            }

            @Override
            public void cancel() {

            }
        });
        Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.type.equals("create");
                    }
                })
                .next("pay")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.type.equals("pay");
                    }
                })
                .within(Time.seconds(5));
        OutputTag<String> outputTag = new OutputTag<String>("late_order") {
        };
        SingleOutputStreamOperator<String> result = CEP
                .pattern(
                        streamSource.keyBy(r -> r.orderId),
                        pattern
                )
                .flatSelect(
                        //侧输出流标签
                        outputTag,
                        //迟到数据  =》输出到测输出流
                        new PatternFlatTimeoutFunction<OrderEvent, String>() {
                            @Override
                            public void timeout(Map<String, List<OrderEvent>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                                //map中只有create 不包含pay
                                out.collect("订单：" + pattern.get("create").get(0).orderId + "未支付");
                            }
                        },
                        //未迟到数据
                        new PatternFlatSelectFunction<OrderEvent, String>() {
                            @Override
                            public void flatSelect(Map<String, List<OrderEvent>> pattern, Collector<String> out) throws Exception {
                                out.collect("订单：" + pattern.get("pay").get(0).orderId + "支付成功");
                            }
                        }
                );
        result.print();
        result.getSideOutput(outputTag).print();

        env.execute();
    }
    public static class OrderEvent{
        public String orderId;
        public String type;
        public Long timestamp;

        @Override
        public String toString() {
            return "OrderEvent{" +
                    "orderId='" + orderId + '\'' +
                    ", type='" + type + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }

        public OrderEvent() {
        }

        public OrderEvent(String orderId, String type, Long timestamp) {
            this.orderId = orderId;
            this.type = type;
            this.timestamp = timestamp;
        }
    }
}
